package com.intct.flink.sql;

import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-07-01 11:38
 */
public class FlinkSQLDemo1 {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 创建临时表
        tEnv.createTemporaryTable("MyTable",
                TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10L)
                .option("fields.f0.kind", "random").build());


        Table myTable = tEnv.from("MyTable");

        // 转换算子

        // 执行SQL
        TableResult tableResult = myTable.execute();

        // 输出
        tableResult.print();
    }
}
